package com.iot.fire.netty;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.iot.fire.common.enums.RequestMapper;
import com.iot.fire.common.util.StringUtils;
import com.iot.fire.domain.DGatewayMessage;
import com.iot.fire.domain.DWaterDeviceMessage;
import com.iot.fire.domain.DWaterMessage;
import com.iot.fire.domain.TDevice;
import com.iot.fire.iot.DataMessage;
import com.iot.fire.iot.Heartbeat;
import com.iot.fire.service.*;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @PackageName: com.iot.fire.tcp
 * @ClassName: NettyServerHandler
 * @Description:
 * @Author: czl
 * @Date: 2025/11/9 17:26
 */
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private final ExecutorService businessThreadPool;
    private final IDGatewayMessageService gatewayMessageService;
    private final IDWaterMessageService waterMessageService;
    private final IDWaterDeviceMessageService waterDeviceMessageService;
    private final ITDeviceService deviceService;
    private final RequestService requestService;

    private static final AtomicInteger connectionCounter = new AtomicInteger(0);
    private static final AtomicLong totalMessagesProcessed = new AtomicLong(0);

    // 存储设备连接信息，key为设备SN码，value为ChannelHandlerContext
    private static final ConcurrentHashMap<String, ChannelHandlerContext> deviceChannels = new ConcurrentHashMap<>();
    // 存储设备连接信息，key为ChannelHandlerContext，value为设备SN码
    private static final ConcurrentHashMap<ChannelHandlerContext, String> channelDevices = new ConcurrentHashMap<>();

    // 用于存储设备SN的属性键
    private static final AttributeKey<String> DEVICE_SN_KEY = AttributeKey.valueOf("deviceSn");

    // 消息处理队列，用于缓冲高并发消息
    private final BlockingQueue<DeviceMessageTask> messageQueue = new LinkedBlockingQueue<>(50000);

    // 消息处理线程
    private final ExecutorService messageProcessorPool;

    private static final List<String> GATEWAY_FAULT_TYPES = Arrays.asList("电源故障", "主电故障", "备电故障", "通讯故障", "故障");
    private static final List<String> GATEWAY_RESET_TYPES = Arrays.asList("复位", "故障恢复", "主电故障恢复", "备电故障恢复", "备电恢复", "恢复", "主电恢复");
    private static final List<String> GATEWAY_OTHER_TYPES = Arrays.asList("消音", "反馈", "消声");
    private static final List<String> GATEWAY_FIRE_TYPES = Arrays.asList("火警", "首次火警", "火灾报警", "火灾后续");


    public NettyServerHandler(ExecutorService businessThreadPool, IDGatewayMessageService gatewayMessageService,
                              IDWaterMessageService waterMessageService, IDWaterDeviceMessageService waterDeviceMessageService,
                              ITDeviceService deviceService, RequestService requestService) {
        this.businessThreadPool = businessThreadPool;
        this.gatewayMessageService = gatewayMessageService;
        this.waterMessageService = waterMessageService;
        this.waterDeviceMessageService = waterDeviceMessageService;
        this.deviceService = deviceService;
        this.requestService = requestService;

        // 创建专门的消息处理线程池
        this.messageProcessorPool = new ThreadPoolExecutor(
                4,
                8,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(50000),
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r, "message-processor-" + threadNumber.getAndIncrement());
                        t.setDaemon(false);
                        t.setPriority(Thread.NORM_PRIORITY);
                        return t;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        // 启动消息处理线程
        startMessageProcessor();
    }

    /**
     * 启动消息处理器线程
     */
    private void startMessageProcessor() {
        // 根据8核服务器调整线程数，建议设置为CPU核心数或稍高
        int processorThreads = Runtime.getRuntime().availableProcessors();

        for (int i = 0; i < processorThreads; i++) {
            messageProcessorPool.submit(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        DeviceMessageTask task = messageQueue.take();
                        processMessageTask(task);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        log.warn("消息处理器线程中断");
                        break;
                    } catch (Exception e) {
                        log.error("错误处理消息任务", e);
                    }
                }
            });
        }
    }

    /**
     * 处理消息任务
     */
    private void processMessageTask(DeviceMessageTask task) {
        try {
            processGBKHexMessage(task.bytes, task.ctx);
            totalMessagesProcessed.incrementAndGet();
        } catch (Exception e) {
            log.error("处理消息任务时发生异常: {}", e.getMessage(), e);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        int currentConnections = connectionCounter.incrementAndGet();
        log.info("新客户端连接: {},，当前连接数:{}", ctx.channel().remoteAddress(), currentConnections);
        super.channelActive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ByteBuf) {
            ByteBuf byteBuf = (ByteBuf) msg;
            byte[] bytes = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(bytes);
            // 释放资源
            byteBuf.release();

            // 将任务提交到队列，避免线程池拒绝
            DeviceMessageTask task = new DeviceMessageTask(bytes, ctx);
            boolean queued = messageQueue.offer(task, 1, TimeUnit.SECONDS);
            if (!queued) {
                log.error("消息队列已满，丢弃消息来自: {}", ctx.channel().remoteAddress());
            }

        } else {
            log.error("收到未知类型的消息: {}", msg.getClass().getName());
        }
    }

    private void processGBKHexMessage(byte[] bytes, ChannelHandlerContext ctx) {
        try {
            // 原始字符串（十六进制字符）
            String originalString = new String(bytes, "GBK").trim();
            log.info("UTF-8解码后的字符串: {}", originalString);

            // 解析并处理消息
            parseAndProcessMessage(originalString, ctx);

            // 检查消息内容，根据条件发送响应
            String responseContent = generateResponse(originalString);
            log.info("发送响应内容: {}", responseContent);

            // 将响应内容转换为GBK编码的字节数组
            byte[] responseBytes = responseContent.getBytes("GBK");

            // 将响应字节数组转换为十六进制字符串
            String hexResponse = bytesToHex(responseBytes).replace(" ", " ");
            log.info("十六进制响应内容: {}", hexResponse);

            // 回复客户端
            ctx.writeAndFlush(hexResponse + "\n");

        } catch (Exception e) {
            log.error("处理GBK十六进制消息失败: {}", e.getMessage());
            e.printStackTrace();
        }
    }

    private String tryDecodeWithMultipleEncodings(byte[] bytes) {
        // 尝试GBK编码
        try {
            String gbkResult = new String(bytes, "GBK");
            if (isValidJsonOrExpectedFormat(gbkResult)) {
                log.debug("使用GBK编码成功解码");
                return gbkResult;
            }
        } catch (Exception e) {
            log.debug("GBK解码失败: {}", e.getMessage());
        }

        // 尝试UTF-8编码
        try {
            String utf8Result = new String(bytes, "UTF-8");
            if (isValidJsonOrExpectedFormat(utf8Result)) {
                log.debug("使用UTF-8编码成功解码");
                return utf8Result;
            }
        } catch (Exception e) {
            log.debug("UTF-8解码失败: {}", e.getMessage());
        }

        // 默认返回GBK结果
        try {
            return new String(bytes, "GBK");
        } catch (Exception e) {
            return new String(bytes); // 使用默认编码
        }
    }

    private boolean isValidJsonOrExpectedFormat(String content) {
        return content.contains("@@") && content.contains("##") &&
                (content.contains("Heartbeat") || content.contains("Data_message"));
    }


    // 将字节数组转换为十六进制字符串
    private static String bytesToHex(byte[] bytes) {
        StringBuilder result = new StringBuilder();
        for (byte b : bytes) {
            result.append(String.format("%02X ", b));
        }
        return result.toString().trim();
    }

    // 根据消息内容生成响应
    private String generateResponse(String decodedMessage) {
        // 检查是否包含Heartbeat
        if (decodedMessage.contains("Heartbeat")) {
            // 如果消息包含Heartbeat，返回指定格式
            return "@@{\n\"DT\":\"Heartbeat\",\n\"US\":\"00001\",\n}3FAA##\n";
        } else if (decodedMessage.contains("Data_message") && decodedMessage.contains("HM")) {
            // 如果消息包含Data_message和HM为网关协议，返回指定格式
            return "@@{\n\"DT\":\"Data_message\",\n\"US\":\"00010\",\n}6C03##\n";
        } else if (decodedMessage.contains("Data_message") && decodedMessage.contains("EM")) {
            // 如果消息包含Data_message和EM为水协议，返回指定格式
            return "@@{\n\"DT\":\"Data_message\",\n\"US\":\"00010\",\n}6C03##\n";
        } else {
            // 如果不包含Heartbeat，返回通用响应
            return "@@{\n\"DT\":\"Response\",\n\"US\":\"00001\",\n}3FAA##\n";
        }
    }

    // 将十六进制字符串转换为字节数组
    private byte[] hexStringToByteArray(String hex) {
        // 只保留有效的十六进制字符
        hex = hex.replaceAll("[^0-9A-Fa-f]", "");

        // 如果长度是奇数，前面补0
        if (hex.length() % 2 != 0) {
            hex = "0" + hex;
        }

        byte[] bytes = new byte[hex.length() / 2];
        for (int i = 0; i < bytes.length; i++) {
            String hexByte = hex.substring(i * 2, i * 2 + 2);
            bytes[i] = (byte) Integer.parseInt(hexByte, 16);
        }
        return bytes;
    }

    private void parseAndProcessMessage(String message, ChannelHandlerContext ctx) {
        try {
            log.info("开始解析消息内容...");

            // 检查是否包含起始标记
            if (message.startsWith("@@")) {
                log.info("✓ 检测到起始标记 @@");
                // 提取JSON部分
                int jsonStart = message.indexOf('{');
                int jsonEnd = message.lastIndexOf('}');

                String jsonContent = null;
                if (jsonStart != -1 && jsonEnd != -1 && jsonEnd > jsonStart) {
                    jsonContent = message.substring(jsonStart, jsonEnd + 1);
                    log.info("✓ 提取的JSON内容: {}", jsonContent);

                    // 解析JSON内容获取设备SN
                    String deviceSN = extractDeviceSN(jsonContent);
                    if (deviceSN != null) {
                        // 存储设备SN到Channel属性
                        ctx.attr(DEVICE_SN_KEY).set(deviceSN);

                        // 注册设备到设备通道映射
                        deviceChannels.put(deviceSN, ctx);
                        channelDevices.put(ctx, deviceSN);

                        log.info("设备 {} 已注册到连接 {}", deviceSN, ctx.channel().remoteAddress());
                    }

                }

                // 检查结束标记
                if (message.contains("##")) {
                    log.info("✓ 检测到结束标记 ##");
                }

                if (jsonContent != null) {
                    // 心跳消息（上行）
                    if (jsonContent.contains("Heartbeat")) {
                        log.info("✓ 检测到心跳消息");
                        // 解析JSON字段
                        Heartbeat heartbeat = JSONUtil.toBean(jsonContent, Heartbeat.class);
                        String info = StrUtil.isBlank(heartbeat.getHM()) ? "水协议" : "网关协议";
                        log.info("{} 心跳消息，心跳内容：{}", info, JSONUtil.toJsonStr(heartbeat));
                    }
                    // 数据消息（上行）
                    if (jsonContent.contains("Data_message")) {
                        DataMessage dataMessage = JSONUtil.toBean(jsonContent, DataMessage.class);
                        if (StrUtil.isBlank(dataMessage.getHM())) {
                            // 水协议数据消息，异步处理
                            businessThreadPool.submit(() -> {
                                try {
                                    log.debug("水协议数据消息，数据内容：{}", JSONUtil.toJsonStr(dataMessage));
                                    DWaterMessage waterMessage = BeanUtil.toBean(dataMessage, DWaterMessage.class);
                                    waterMessage.setTime(new Date());

                                    TDevice deviceInfo = deviceService.getDeviceByDeviceSn(waterMessage.getSN());
                                    if (null == deviceInfo) return;

                                    DWaterMessage lastWaterMessageByDeviceSn = waterMessageService.getLastWaterMessageByDeviceSn(waterMessage.getSN());
                                    boolean shouldSave = false;

                                    if (lastWaterMessageByDeviceSn == null) {
                                        log.warn("未找到设备{}的历史消息记录", waterMessage.getSN());
                                        shouldSave = true;
                                    } else {
                                        Date lastTime = lastWaterMessageByDeviceSn.getTime();
                                        long timeDifferenceInMillis = System.currentTimeMillis() - lastTime.getTime();
                                        if (timeDifferenceInMillis > TimeUnit.MINUTES.toMillis(30)) {
                                            shouldSave = true;
                                        }
                                    }

                                    if (shouldSave) {
                                        boolean saveStatus = waterMessageService.save(waterMessage);
                                        if (saveStatus) {
                                            List<DWaterDeviceMessage> waterDeviceMessages = new ArrayList<>();

                                            processField(dataMessage.getA1(), "A1", waterMessage.getId(), waterDeviceMessages);
                                            processField(dataMessage.getA2(), "A2", waterMessage.getId(), waterDeviceMessages);
                                            processField(dataMessage.getA3(), "A3", waterMessage.getId(), waterDeviceMessages);

                                            if (!waterDeviceMessages.isEmpty()) {
                                                waterDeviceMessageService.saveBatch(waterDeviceMessages);
                                            }
                                        }
                                    }


                                } catch (Exception e) {
                                    log.error("处理水协议数据消息失败", e);
                                }
                            });
                        } else {
                            // 网关协议数据消息，异步处理
                            businessThreadPool.submit(() -> {
                                try {
                                    log.debug("网关协议数据消息，数据内容：{}", JSONUtil.toJsonStr(dataMessage));
                                    DGatewayMessage gatewayMessage = BeanUtil.toBean(dataMessage, DGatewayMessage.class);
                                    gatewayMessage.setTime(new Date());

                                    TDevice deviceInfo = deviceService.getDeviceByDeviceSn(gatewayMessage.getSN());
                                    if (null == deviceInfo) return;

                                    // 网关设备数据上报
                                    handleGatewayDevice(deviceInfo, gatewayMessage);

                                    gatewayMessageService.save(gatewayMessage);
                                } catch (Exception e) {
                                    log.error("处理网关协议数据消息失败", e);
                                }
                            });
                        }
                    }
                }
            }

        } catch (Exception e) {
            log.error("解析消息失败: {}", e.getMessage());
        }
    }

    /**
     * 处理网关设备数据上报
     *
     * @param deviceInfo 设备信息
     */
    private void handleGatewayDevice(TDevice deviceInfo, DGatewayMessage gatewayMessage) {
        // 网关协议
        Map<String, Object> params = new HashMap<>();
        // 设备id,创建设备时获得
        params.put("device_id", deviceInfo.getDeviceId());

        // 报警状态，0-正常，1-报警  PS是火警是报警  其他为正常
        if (StringUtils.containsAny(GATEWAY_FIRE_TYPES, gatewayMessage.getPS())) {
            params.put("alarm_status", 1);
        } else {
            params.put("alarm_status", 0);
        }

        // 故障状态，0-正常，1-故障 PS是故障是故障 其他为正常
        if (StringUtils.containsAny(GATEWAY_FAULT_TYPES, gatewayMessage.getPS())) {
            params.put("abnormal_status", 1);
        } else {
            params.put("abnormal_status", 0);
        }

        // 屏蔽状态，0-正常，1-异常 都是正常
        params.put("shield_status", 0);
        // 事件发生时间，毫秒级时间戳，格式 1539776909000
        params.put("reported_at", System.currentTimeMillis());

        Map<String, Object> childData = new HashMap<>();

//        String ps = gatewayMessage.getPS();
//        if (StringUtils.containsAny(GATEWAY_FAULT_TYPES, ps)) {
//            childData.put("data_type_id", 15);
//        } else if (StringUtils.containsAny(GATEWAY_RESET_TYPES, ps)) {
//            childData.put("data_type_id", 16);
//        } else if (StringUtils.containsAny(GATEWAY_OTHER_TYPES, ps)) {
//            childData.put("data_type_id", 42);
//        } else if (StringUtils.containsAny(GATEWAY_FIRE_TYPES, ps)) {
//            childData.put("data_type_id", 84);
//        }
        childData.put("data_type_id", 42);

        // data_value PT的值
        childData.put("data_value", gatewayMessage.getPS());
        // alarm_status PS 为火警为报警 其他为正常
        if (StringUtils.containsAny(GATEWAY_FIRE_TYPES, gatewayMessage.getPS())) {
            childData.put("alarm_status", 1);
        } else {
            childData.put("alarm_status", 0);
        }

        // abnormal_status PS为故障为故障 其他为正常
        if (StringUtils.containsAny(GATEWAY_FAULT_TYPES, gatewayMessage.getPS())) {
            childData.put("abnormal_status", 1);
        } else {
            childData.put("abnormal_status", 0);
        }
        // threshold_const_value 阈值 0
        childData.put("threshold_const_value", "");
        childData.put("threshold_max_value", "");
        childData.put("threshold_min_value", "");

        List<Map<String, Object>> dataList = new ArrayList<>(1);
        dataList.add(childData);

        params.put("data", JSON.toJSONString(dataList));

        requestService.advancedRequest(RequestMapper.DYNAMIC_DATA_UPLOAD, params);
    }

    // 提取字段处理逻辑
    private void processField(String field, String device, String waterId, List<DWaterDeviceMessage> messages) {
        if (StrUtil.isBlank(field)) {
            return;
        }

        String[] parts = field.split("-");
        if (parts.length != 6) {
            log.warn("字段格式不正确: {}", field);
            return;
        }

        DWaterDeviceMessage msg = new DWaterDeviceMessage();
        msg.setWaterId(waterId);
        msg.setDevice(device);
        msg.setField1(parts[0]);
        msg.setField2(parts[1]);
        msg.setField3(parts[2]);
        msg.setField4(parts[3]);
        msg.setField5(parts[4]);
        msg.setField6(parts[5]);

        messages.add(msg);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof io.netty.handler.timeout.IdleStateEvent) {
            log.warn("检测到连接空闲，关闭连接: {}", ctx.channel().remoteAddress());

            // 原子操作减少连接数
            connectionCounter.getAndDecrement();

            // 从设备映射中移除
            String deviceSN = ctx.attr(DEVICE_SN_KEY).get();
            if (deviceSN != null) {
                deviceChannels.remove(deviceSN);
                channelDevices.remove(ctx);
                log.info("设备 {} 从连接映射中移除", deviceSN);
            }

            ctx.close();
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("处理客户端 {} 时发生异常: {}", ctx.channel().remoteAddress(), cause.getMessage());

        // 原子操作减少连接数
        connectionCounter.getAndDecrement();

        // 从设备映射中移除
        String deviceSN = ctx.attr(DEVICE_SN_KEY).get();
        if (deviceSN != null) {
            deviceChannels.remove(deviceSN);
            channelDevices.remove(ctx);
            log.info("设备 {} 从连接映射中移除", deviceSN);
        }

        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 原子操作减少连接数
        int finalCount = connectionCounter.decrementAndGet();

        // 从设备映射中移除
        String deviceSN = ctx.attr(DEVICE_SN_KEY).get();
        if (deviceSN != null) {
            deviceChannels.remove(deviceSN);
            channelDevices.remove(ctx);
            log.info("设备 {} 从连接映射中移除", deviceSN);
        }

        log.info("客户端断开连接: {}，当前连接数: {}", ctx.channel().remoteAddress(), finalCount);
        super.channelInactive(ctx);
    }

    // 从JSON内容中提取设备SN
    private String extractDeviceSN(String jsonContent) {
        try {
            // 从JSON中提取SN字段
            if (jsonContent.contains("\"SN\":\"")) {
                int snStart = jsonContent.indexOf("\"SN\":\"") + 6;
                int snEnd = jsonContent.indexOf("\"", snStart);
                if (snStart > 5 && snEnd > snStart) {
                    return jsonContent.substring(snStart, snEnd);
                }
            }
        } catch (Exception e) {
            log.error("提取设备SN失败: {}", e.getMessage());
        }
        return null;
    }

    // 通过设备SN向设备发送消息
    public static boolean sendToDevice(String deviceSN, String message) {
        ChannelHandlerContext ctx = deviceChannels.get(deviceSN);
        if (ctx != null && ctx.channel().isActive()) {
            try {
                // 将消息转换为GBK编码的字节数组
                byte[] messageBytes = message.getBytes("GBK");

                // 将字节数组转换为十六进制字符串
                String hexMessage = bytesToHex(messageBytes);

                log.info("向设备 {} 发送消息: {}", deviceSN, hexMessage);

                // 发送消息
                ctx.writeAndFlush(hexMessage + "\n");
                return true;
            } catch (Exception e) {
                log.error("向设备 {} 发送消息失败: {}", deviceSN, e.getMessage());
                return false;
            }
        } else {
            log.warn("设备 {} 未连接或连接已断开", deviceSN);
            return false;
        }
    }

    // 通过设备SN向设备发送十六进制消息
    public static boolean sendHexToDevice(String deviceSN, String hexMessage) {
        ChannelHandlerContext ctx = deviceChannels.get(deviceSN);
        if (ctx != null && ctx.channel().isActive()) {
            log.info("向设备 {} 发送十六进制消息: {}", deviceSN, hexMessage);

            // 发送十六进制消息
            ctx.writeAndFlush(hexMessage + "\n");
            return true;
        } else {
            log.warn("设备 {} 未连接或连接已断开", deviceSN);
            return false;
        }
    }

    // 获取设备连接状态
    public static boolean isDeviceConnected(String deviceSN) {
        ChannelHandlerContext ctx = deviceChannels.get(deviceSN);
        return ctx != null && ctx.channel().isActive();
    }

    // 获取连接设备数量
    public static int getConnectedDeviceCount() {
        return deviceChannels.size();
    }

    // 获取所有连接的设备SN
    public static String[] getConnectedDeviceSNs() {
        return deviceChannels.keySet().toArray(new String[0]);
    }

    // 获取总处理消息数
    public static long getTotalMessagesProcessed() {
        return totalMessagesProcessed.get();
    }

    /**
     * 设备消息任务类
     */
    private static class DeviceMessageTask {
        final byte[] bytes;
        final ChannelHandlerContext ctx;

        DeviceMessageTask(byte[] bytes, ChannelHandlerContext ctx) {
            this.bytes = bytes;
            this.ctx = ctx;
        }
    }
}
